package com.yuanhua.sim.gateway.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.spring.util.BeanUtils;
import com.yuanhua.core.base.R;
import com.yuanhua.core.base.entity.MsgReqBO;
import com.yuanhua.core.base.entity.MsgReqBean;
import com.yuanhua.core.exception.BaseException;
import com.yuanhua.core.exception.BizException;
import com.yuanhua.core.exception.code.BaseExceptionCode;
import com.yuanhua.core.exception.code.ExceptionCode;
import com.yuanhua.sim.gateway.contants.MQTopic;
import com.yuanhua.sim.gateway.model.PathMsg;
import com.yuanhua.sim.gateway.service.GateWayService;
import com.yuanhua.sim.gateway.utils.IncriAndUniqueID;
import com.yuanhua.sim.gateway.utils.RocketMQUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 处理消息发送及很多业务
 * @author yuanhua
 */
@Service
@Slf4j
public class GateWayServiceImpl implements GateWayService {
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private RocketMQUtils rocketMQUtils;

    @Autowired
    private ZooKeeper zooKeeper;

    private AtomicInteger state = new AtomicInteger(0);

    List<String> children = new ArrayList<>(4);
    /**
     * 消息发送业务处理
     * @param msg
     * @return
     */
    @Override
    public String sendMsg(MsgReqBO msg) {
        //1判断消息是否重复---redis
        String id = msg.getMid();
        redisTemplate.hasKey(id);
        if (Boolean.TRUE.equals(redisTemplate.hasKey(id))){
            log.info("消息重复---{}",msg.getMid());
            //消息重复
            return redisTemplate.boundValueOps(id).get();
        }
        //没重复，生成唯一ID发送构建消息体，
        String mid = IncriAndUniqueID.getIncriAndUniqueId(0, 1);
        msg.setMid(mid);
        //判断接收人是否在线----redis
        msgSendOrCache(msg);
        long start = System.currentTimeMillis();
        //异步发送到MQ；发送异常进行重发 发送失败会进行重发；重试次数：retryTimesWhenSendFailed: 4
        try {
            msg.setCreateTime(LocalDateTime.now());
            rocketMQUtils.messageSend(MQTopic.ON_LINE_TOPIC, msg);
        } catch (Exception e) {
            //重发结束依旧失败；
            e.printStackTrace();
            //返回发送失败， 服务器繁忙~~~；
            throw BizException.wrap(ExceptionCode.SYSTEM_BUSY);
        }
        System.err.println("消息发送耗时----" + (System.currentTimeMillis() - start) + "毫秒");

        //收到响应，返回消息ID给前端，前端收到消息ID就发送成功
        //巧妙将客户端传过来的的id保存并设置成已经发送的ID，方便下次重复直接返回
        redisTemplate.boundValueOps(id).set(mid);
        return mid;
    }

    /**
     * 消息发送或缓存
     * @param:
     * @param msg 消息
     * @return: void
     * @author songguoxiang
     * @date: 2022/8/15 15:31
     */
    private void msgSendOrCache(MsgReqBO msg) {
        if(Boolean.TRUE.equals(redisTemplate.hasKey(msg.getRecverId().toString()))){
            //在线直接转发到该netty服务上
            //获取用户所在服务器
            String  path = redisTemplate.boundValueOps(msg.getRecverId().toString()).get();
            if (StringUtils.hasText(path)) {
                //服务下线了！需要删除该服务对应的关系！

            }
            //redis中用户及服务 数据格式：（hash） userID - ip:port
            String[] ipAndPort = path.split(":");
            //调用发送
            MsgReqBean msgReqBean = new MsgReqBean();
            BeanUtil.copyProperties(msg, msgReqBean);
            boolean b = sendOnlineMsg(ipAndPort[0], ipAndPort[2],  msgReqBean);
            if (!b) {
                //删除 C - S键值对
            }
        } else {
            //不在线，需要存储消息到redis;
            //异步将消息发送到MQ，由离线服务消费并保存到redis中
            msg.setCreateTime(LocalDateTime.now());
            Boolean dfas = redisTemplate.boundZSetOps("45:59").add("dfas", 45);
            //rocketMQUtils.messageSend(MQTopic.IM_CACHE_TOPIC, msg);

        }
    }

    /**
     * 算法挑选服务返回
     *
     * @return
     */
    @Override
    public PathMsg getPath() throws KeeperException, InterruptedException {
        if (children.isEmpty()) {
            children = zooKeeper.getChildren("/route", true);
        }
        //如果恰好服务发生变化，要等变化结束后方可获取
        while (true) {
            if (state.get() == 0) {
                break;
            }else {
                wait();
            }
        }
        if (CollUtil.isEmpty(children)){
            return null;
        }
        log.info("NettyServer服务列表：{}", children);
        Random random = new Random();
        int idx = random.nextInt(children.size());
        String s = children.get(idx);
        String[] split = s.split("-");
        String[] res = split[1].split(":");
        return new PathMsg(res[0], res[1]);
    }

    /**
     * 服务节点检测
     *
     * @author songguoxiang
     * @date: 2022/8/15 11:28
     */
    @Override
    public void doReGetChildren() throws InterruptedException, KeeperException {
        state.incrementAndGet();
        children = zooKeeper.getChildren("/route", true);
        state.decrementAndGet();
        notify();
        log.info("服务状态发生变化---{}",children);
    }


    private boolean sendOnlineMsg(String ip, String port, MsgReqBean msgReqBean) {
        String url = "http://" + ip + ":" + port + "/netty/send";
        R res = restTemplate.postForObject(url, msgReqBean, R.class);
        //如果发送失败，说明该用户不在线了！
        if (res != null && res.getIsError()){
            log.info("用户已经不在线了");
            return false;
        }
        return true;
    }
}
